package com.rech.redis;

import android.os.Message;
import android.util.Log;

import com.rech.utils.LogUtil;
import com.rech.utils.OtherUtil;
import com.rech.xbexpress3.ExpressApp;
import com.rech.xbexpress3.RedisLogic;
import com.rech.xbexpress3.RedisWorker;

import org.greenrobot.eventbus.EventBus;

import java.net.SocketException;
import java.net.SocketTimeoutException;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.exceptions.JedisConnectionException;

import static com.rech.xbexpress3.RedisWorker.jedis;

/**
 * Created by sunshine on 2020/7/20.
 */
public class MessageConsumer implements Runnable {
    public static final String CHANNEL_KEY = "100001";//频道

    public static final String EXIT_COMMAND = "exit";//结束程序的消息

    private MyJedisPubSub myJedisPubSub = new MyJedisPubSub();//处理接收消息

    private int num;

    public void consumerMessage() {
        try {
            jedis.subscribe(myJedisPubSub, CHANNEL_KEY);//第一个参数是处理接收消息，第二个参数是订阅的消息频道
            LogUtil.getInstance().info("redis订阅/连接成功：");
            Log.e("redis日志", "redis订阅/连接成功：");
        } catch (JedisConnectionException e) {
            Message msg = new Message();
            msg.what = 9999;
            msg.obj = "心跳异常" + OtherUtil.getTime() + e.toString();
            EventBus.getDefault().post(msg);

            LogUtil.getInstance().info("redis订阅/连接错误：" + e.getMessage());
            Log.e("redis日志", "redis订阅/连接错误：" + e.getMessage());

            jedis.close();
            try {
                Thread.sleep(10000);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }

            reconnect();
                /*while (!jedis.isConnected() && OtherUtil.isNetConnected(ExpressApp.getInstance())){
                    jedis.connect();

                    msg.what = 9999;
                    msg.obj = OtherUtil.getTime() + "重连第" + num + "次";
                    num = num++;
                    EventBus.getDefault().post(msg);

                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }*/

            consumerMessage();
        }
    }

    private void reconnect() {
        if (OtherUtil.isNetConnected(ExpressApp.getInstance())) {
            try {
                jedis.connect();
            } catch (Exception ex) {
                LogUtil.getInstance().info("redis重新连接错误：" + ex.getMessage());
                Log.e("redis日志", "redis重新连接错误：" + ex.getMessage());
                ex.printStackTrace();

                Message msg = new Message();
                msg.what = 9999;
                msg.obj = OtherUtil.getTime() + "尝试重连";
                num = num++;
                EventBus.getDefault().post(msg);

                jedis.close();
                reconnect();
            }
        }
    }

    @Override
    public void run() {
        consumerMessage();
        while (true) {
            consumerMessage();

            new Thread(new Runnable() {
                @Override
                public void run() {
                    //while (!jedis.isConnected() && OtherUtil.isNetConnected(ExpressApp.getInstance()) && OtherUtil.netCanUse(ExpressApp.getInstance())) {
                    while (true) {
                        String result = jedis.ping();
                        if (null == result || !result.equals("PONG")) {
                            LogUtil.getInstance().info("redis进行PING时错误" );
                            Log.e("redis日志", "redis进行PING时错误");

                            jedis.close();
                            jedis.connect();

                            Message msg = new Message();
                            msg.what = 9999;
                            msg.obj = OtherUtil.getTime() + "网络不可用重连" + result.toString();
                            EventBus.getDefault().post(msg);

                            try {
                                Thread.sleep(10000);
                            } catch (InterruptedException ex) {
                                ex.printStackTrace();
                            }

                            consumerMessage();
                        } else {
                            LogUtil.getInstance().info("redis进行PING时正确" );
                            Log.e("redis日志", "redis进行PING时正确");
                        }
                    }
                }
            }).start();
        }
    }
}

/**
 * 继承JedisPubSub，重写接收消息的方法
 */
class MyJedisPubSub extends JedisPubSub {
    @Override
    /** JedisPubSub类是一个没有抽象方法的抽象类,里面方法都是一些空实现
     * 所以可以选择需要的方法覆盖,这儿使用的是SUBSCRIBE指令，所以覆盖了onMessage
     * 如果使用PSUBSCRIBE指令，则覆盖onPMessage方法
     * 当然也可以选择BinaryJedisPubSub,同样是抽象类，但方法参数为byte[]
     **/
    public void onMessage(String channel, String message) {
        System.out.println(Thread.currentThread().getName()+"-接收到消息:channel=" + channel + ",message=" + message);
        //接收到exit消息后退出
        if (MessageConsumer.EXIT_COMMAND.equals(message)) {
            System.exit(0);

            Message msg = new Message();
            msg.what = 9999;
            msg.obj = "程序结束" + OtherUtil.getTime();
        }

        try {
            if (message != null) {
                //recvBuffer.add(result);
                RedisLogic.getInstance().doWorker(message);
                Message msg = new Message();
                msg.what = 9999;
                msg.obj = "正常" + OtherUtil.getTime();
                EventBus.getDefault().post(msg);
            } else {
                Message msg = new Message();
                msg.what = 9999;
                msg.obj = "为空" + OtherUtil.getTime();
                EventBus.getDefault().post(msg);
            }
        } catch (Exception e) {
            jedis.close();
            Message msg = new Message();
            msg.what = 9999;
            msg.obj = "异常" + OtherUtil.getTime() + e.toString();
            EventBus.getDefault().post(msg);
        }
    }
}
